package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/12 23:54
 */
public class Flink16_UDTF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });


        Table table = tableEnv.fromDataStream(sensorDS);


        // 先注册
        tableEnv.createTemporaryFunction("split", Split.class);

        // TableAPI
//        table
//                .joinLateral(call("split",$("id")))
//                .select($("id"),$("word"))
//                .execute()
//                .print();


        // SQL,已经在自定义udtf的时候，指定了hint，炸开字段叫 word
        tableEnv.sqlQuery("select id,word from " + table + ",lateral table(split(id))").execute().print();

        // SQL,没指定hint，单独起个别名, as 表名(字段名)，这样就可以在 select中使用
//        tableEnv.sqlQuery("select id,word from " + table + ",lateral table(split(id)) as tmp(word)").execute().print();

    }

    @FunctionHint(output = @DataTypeHint("ROW<word STRING>"))
    public static class Split extends TableFunction<Row> {
        public void eval(String value) {
            String[] split = value.split("_");
            for (String s : split) {
                collect(Row.of(s));
            }
        }
    }
}
